package gbench.appdemo.priceline.app;

import gbench.appdemo.priceline.calendar.ITradingCalendar;
import gbench.appdemo.priceline.calendar.TradingCalendar;
import gbench.common.tree.LittleTree.IRecord;
import gbench.common.tree.LittleTree.Tuple2;
import gbench.commonApp.data.DataApp;
import gbench.appdemo.priceline.calendar.TradingCalendar.TimeSpan;

import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.BiFunction;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Predicate;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import static gbench.common.tree.LittleTree.CronTime.*;
import static gbench.common.tree.LittleTree.IRecord.*;

/**
 * 
 * @author gbench
 *
 */
public class PriceLineApplication {

    /**
     * 构造哈桑农户
     * @param ticker 合约代码
     * @param period 价格线的周期
     * @param callback 回调函数
     */
    public PriceLineApplication(final String ticker,final Long period,final Consumer<IRecord> callback){
        
        this.ticker = ticker;
        this.period = period;
        this.calendar = mycalendar(ticker);
        this.appEngine = new PriceLineAppEngine();// 价格线计算引擎
        this.callback = callback;
    }
    
    /**
     * 构造哈桑农户
     * @param ticker 合约代码
     * @param period 价格线的周期
     * @param pseudo_processor 引擎处理器
     * @param callback 回调函数
     */
    public PriceLineApplication(final String ticker,final Long period,
        final BiFunction<PriceLineAppEngine,IRecord,IRecord>pseudo_processor,final Consumer<IRecord> callback){
        
        this.ticker = ticker;
        this.period = period;
        this.calendar = mycalendar(ticker);
        this.appEngine = new PriceLineAppEngine(pseudo_processor);// 价格线计算引擎
        this.callback = callback;
    }
    
    /**
     * 数据应用初始化
     * @param appletBuilder
     */
    public void initialize(final Function<PriceLineApplication,DataApplet> appletBuilder) {
        final var applet = appletBuilder.apply(this);
        applet.execute(appEngine::handle);
    }
    
    /**
     * 把一个IRecord记录转换成Calendar结构。  <br>
     *   // 交易日历 <br>
     *   var rec = REC( <br>
     *       "name","MA001","ticker","MA001","description","MA001", // name:日历名称,ticker:合约代码,description:日历说明 <br>
     *       "start",ld("2019-01-01"),"end",ld("2019-12-31"), // 开始时间:包含,结束时间:不包含 <br>
     *       "excludes",IRecord.L("2019-01-01,2019-01-05,2019-09-13,2019-09-30,2019-10-01,2019-10-02,2019-10-03,2019-10-04".split(",")) <br>
     *           .stream().map(e->ld(e)).collect(Collectors.toList()), // 排除在外的日期 <br>
     *       "timespans",IRecord.L(// 时间段:单日内的时间跨度 <br>
     *               R("name","上午-前","start",lt("09:00"),"end",lt("10:15")), <br>
     *               R("name","上午-后","start",lt("10:30"),"end",lt("11:30")), <br>
     *               R("name","下午","start",lt("13:30"),"end",lt("15:00")), <br>
     *               R("name","夜间","start",lt("21:00"),"end",lt("23:00")) <br>
     *        ),// timespan <br>
     *       "createtime",now() <br>
     *   );// 交易日历的定义
     * 
     * @param r 一个日历记录结构 见上图示例
     * @return SimpleTradingCalendar
     */
    public static ITradingCalendar rec2calendar(IRecord r) {
        final var calendar = new TradingCalendar(r.str("ticker"),r.ld("start"),r.ld("end"));
        //非法日期
        r.lla("excludes",e->REC("_k",e).ld("_k")).forEach(calendar::addExclude);
        //交易时段
        r.lla("timespans",e->(IRecord)e).stream().map(e->
            new TimeSpan(e.str("name"),e.lt("start"),e.lt("end")))
        .forEach(calendar::addSpan);
        return calendar;
    }
    
    /**
     * 交易日历
     * @return
     */
    public static ITradingCalendar mycalendar(String ticker) {
        final var rec = REC(// 交易日历的配置
            "name",ticker,"ticker",ticker,"description",ticker, // 交易日历的名称
            "start",ld("2020-01-01"),"end",ld("2021-01-01"),// 起止时间
            "excludes",IRecord.L("2020-03-19".split(",")) // 例外日期
                .stream().map(e->ld(e)).collect(Collectors.toList()),
            "timespans",IRecord.L(// 交易时段
                R("name","上午-前","start",lt("09:00"),"end",lt("10:15")),
                R("name","上午-后","start",lt("10:30"),"end",lt("11:30")),
                R("name","下午","start",lt("13:30"),"end",lt("15:00")),
                R("name","夜间","start",lt("21:00"),"end",lt("23:00"))
             ),// timespan
            "createtime",now()
        );// REC 日历的配置
        return TradingCalendar.rec2calendar(rec);
    }
    
   /**
    * 交易日历
    * @param name 日历名称
    * @param start 开始时间包含 
    * @param end 结束时间不包含
    * @param excludes 剔除的时间间隔
    * @param timespans 每日的交易时段
    * @return mycalendar
    */
    public static MyCalendar mycalendar(
        final String name,
        final LocalDateTime start,final LocalDateTime end,
        final List<Tuple2<LocalDateTime,LocalDateTime>>excludes,
        final List<Tuple2<LocalDateTime,LocalDateTime>>timespans ) {
     
        return new MyCalendar(name, end, end, timespans, timespans);
    }
    
    /**
     * 
     * @author gbench
     *
     */
    public static class MyCalendar{
        public MyCalendar(String name, LocalDateTime start, LocalDateTime end,
                List<Tuple2<LocalDateTime, LocalDateTime>> excludes,
                List<Tuple2<LocalDateTime, LocalDateTime>> timespans) {
            super();
            this.name = name;
            this.start = start;
            this.end = end;
            this.excludes = excludes;
            this.timespans = timespans;
            this.initialize();
        }
        
        private final String name;
        private final LocalDateTime start;
        private final LocalDateTime end;
        private final List<Tuple2<LocalDateTime,LocalDateTime>>excludes;
        private final List<Tuple2<LocalDateTime,LocalDateTime>>timespans;
        
        public void initialize() {
            Collections.sort(excludes, (a,b)->a._1().compareTo(b._1()));
            Collections.sort(timespans, (a,b)->a._1().compareTo(b._1()));
        }
        
        
        /**
         * 文本格式化
         */
        public String toString() {
            return this.name+this.start+this.end;
        }
    }
    
    /**
     * 数据数据库
     * @author gbench
     *
     */
    public class ShfexDataApp extends DataApp{
        
        /**
         * 数据APP
         */
       public ShfexDataApp(){
           super();
           cfg.set("url", "jdbc:mysql://localhost:3306/trading?useSSL=false&serverTimezone=GMT%2B8&allowPublicKeyRetrieval=true");
           this.reload();
        }
        
        /**
         * 数据读取
         * @param line_reader 单行记录处理器
         * @param sql sql语句
         */
        public List<IRecord> forEach(Consumer<IRecord> line_reader,final String sql) {
            final var ar = new AtomicReference<List<IRecord>>();
            @SuppressWarnings("unused")
            final var result = jdbc.withTransaction(sess->{
                final var records = sess.sql2records(sql);
                ar.set(records);
            });
            // 结果回调
            if(line_reader!=null && ar.get()!=null) {
                ar.get().parallelStream().forEach(line_reader);
            }//if
            
            //System.out.println(result);
            return ar.get();
        }
        
        /**
         * 数据读取
         * @param sql sql语句
         */
        public List<IRecord> sqlQuery(String sql) {
           return this.forEach(null, sql);
        }
        
        /**
         * 查询一个结果记录
         * @param sql sql语句
         */
        public IRecord sqlOne(String sql) {
            final var ll = this.forEach(null, sql);
            return ll.size()>0?ll.get(0):null;
        }
        
        /**
         * 查询一个结果记录
         * @param sql sql语句
         */
        public <T> T sqlGet(String sql,Class<T>targetClass) {
            final var ll = this.forEach(null, sql);
            return ll.size()>0?ll.get(0).toTarget(targetClass):null;
        }
        
        /**
         * 数据处理器
         * @param line_handler 数据变换器
         * @param sql sql 语句
         * @return DataApplet
         */
        public DataApplet getApplet(final Function<IRecord,IRecord> line_mapper,String sql){
            return line_reader -> {
                ShfexDataApp.this.forEach(rec->line_reader.accept(line_mapper.apply(rec)), sql);
            };//run
        }// getApplet
    }
    
    /**
     * 简单的数据引用
     * @author gbench
     *
     */
    public interface DataApplet{
        /**
         * 数据应用的启动函数
         */
       public void execute(Consumer<IRecord>line_reader);
    }
    
    /**
     * 价格信息的数据库
     * @author gbench
     *
     */
    public class PriceLineDatabase extends LinkedHashMap<Long,IRecord>{
        
        /**
         * dataStream
         * @return 数据流
         */
        public Stream<IRecord> dataStream(Predicate<? super Map.Entry<Long,IRecord>>predicate){
            return this.entrySet().parallelStream()
                .filter(predicate)
                .sorted((a,b)->((Number)(a.getKey()-b.getKey())).intValue()) // 对数据按照键名进行排血
                .map(e->e.getValue());
        }

        /**
         * dataStream
         * @return 数据流
         */
        public Stream<IRecord> dataStream(){
            return this.dataStream(e->true);
        }
        
        /**
         * 添加一个数据记录
         * @param id 记录id
         * @param value 记录值
         * @return PriceLineDatabase 价格数据库
         */
        public PriceLineDatabase addOne(Number id,IRecord value) {
            this.put(id.longValue(), value);
            return this;
        }
        
        /**
         * 查抄一个数据记录
         * @param id 记录id
         * @return 数据记录
         */
        public IRecord findOne(Number id) {
            return this.get(id.longValue());
        }
        
        /**
         * 是否包含id为主键数据记录
         * @param id 记录id
         * @return 包含返回true，否则返回false
         */
        public boolean has(Number id) {
            return this.containsKey(id);
        }
        
        /**
         * serialVersionUID
         */
        private static final long serialVersionUID = 1L;
        
    }
    
    /**
     * 价格线应用引擎
     * @author gbench
     *
     */
    public class PriceLineAppEngine {
        
        /**
         * App Engine 的构造函数
         */
        public PriceLineAppEngine(){
            priceLineDatabase = new PriceLineDatabase();// 价格线数据
            processor = rec->{
                final var index = rec.lng("index");// 提取索引
                final var price = rec.dbl("price");// 提取价格因素
                final var volume = rec.dbl("volume");// 提取 volume
                final var line = priceLineDatabase.findOne(index);// 提取索引坐标
                if(line!=null) {// update the priceLineDatabase
                    line.compute("high", (Double high)->high>price?high:price);
                    line.compute("low", (Double low)->low>price?price:low);
                    line.set("close",price);
                    line.set("volume",rec.dbl("volume")-line.dbl("_vol"));
                    line.set("time", rec.get("time"));// 设置K线的时间
                } else {// 新的价格线数据
                    priceLineDatabase.addOne(index, REC("index",index,"open",price,"high",price,
                        "low",price,"close",price,"volume",0,"_vol",volume));
                }//if line
                
                return priceLineDatabase.get(index);// 处理器
            };// processor
        }
        
        /**
         * App Engine 的构造函数
         * @param pseudo_processor 伪处理器
         */
        public PriceLineAppEngine(final BiFunction<PriceLineAppEngine,IRecord,IRecord>pseudo_processor){
            priceLineDatabase = new PriceLineDatabase();// 价格线数据
            processor = rec->pseudo_processor.apply(this, rec);
        }
        
        /**
         * 处理数据记录
         * @param rec
         */
        public void handle(IRecord rec) {
            final var ohlc = processor.process(rec);// 处理器的处理
            if(PriceLineApplication.this.callback!=null)PriceLineApplication.this.callback.accept(ohlc);
        }
        
        /**
         * 获取价格线条
         * @param from 开始位置
         */
        public Stream<IRecord> getPriceLine(final Number from) {
            final var al = new AtomicLong(0l);// 计数器
            final var start = from==null?0l:from.longValue();
            return this.priceLineDatabase.dataStream(e->al.getAndIncrement()>=start);
        }
        
        /**
         * 数据库
         */
        private final PriceLineDatabase priceLineDatabase;// 价格线数据
        
        /**
         * ohlc 的K线计算器
         */
        private final Processor<IRecord,IRecord> processor;// 引擎的处理器
        
    }
    
    public interface Processor<T,U>{
        U process(T t);
    }
    
    /**
     * 获取价格线条
     * @param from 开始位置
     */
    public Stream<IRecord> getPriceLine(final Number from) {
        return this.appEngine.getPriceLine(from);
    }
    
    public String getTicker() {
        return ticker;
    }

    public Long getPeriod() {
        return period;
    }
    
    public ITradingCalendar getCalendar() {
        return calendar;
    }

    public PriceLineAppEngine getAppEngine() {
        return appEngine;
    }

    private final String ticker;// 股票代码
    private final Long period;// 周期间隔
    private final PriceLineAppEngine appEngine;
    private final ITradingCalendar calendar;// 交易日历
    private final Consumer<IRecord> callback;
    
}
